
#if 01
#include <stdint.h>
#include <stdlib.h>
#include <unistd.h>
#include <malloc.h>
#include <pthread.h>
#define USE_SELF_SPIN 0

#if USE_SELF_SPIN
#include "spinlock.h"
#else
#include <stdatomic.h>
#endif
#include "threadpool.h"


typedef struct task_s {
    void *next;
    handler_ptr func;
    void *arg;
} task_t;


#if USE_SELF_SPIN
    typedef struct spinlock spinlock_t;
#endif

typedef struct task_queue_s {
    void *head;
    void **tail;
    int block;
#if USE_SELF_SPIN
    spinlock_t lock;
#else
    pthread_spinlock_t lock;
#endif
    pthread_mutex_t mutex;
    pthread_cond_t cond;
} task_queue_t;

struct threadpool_s {
    task_queue_t *task_queue;
    atomic_int quit;
    int thread_count;
    pthread_t *threads;
};


static task_queue_t *_taskqueue_create() {
    LOG("Thread: %lu, Start Function: %s\n", pthread_self(), __func__);
    task_queue_t *queue = (task_queue_t *)malloc(sizeof(task_queue_t));
    if (queue) {
        if (0 == pthread_mutex_init(&queue->mutex, NULL)) {
            if (0 == pthread_cond_init(&queue->cond, NULL)) {
#if USE_SELF_SPIN
                spinlock_init(&queue->lock);
                queue->head = NULL;
                queue->tail = &queue->head;
                queue->block = 1;
                return queue;
#else
                if (0 == pthread_spin_init(&queue->lock, PTHREAD_PROCESS_PRIVATE)) {
                    queue->head = NULL;
                    queue->tail = &queue->head;
                    queue->block = 1;
                    return queue;
                }
#endif
                pthread_cond_destroy(&queue->cond);
            }
            pthread_mutex_destroy(&queue->mutex);
        }
        free(queue);
    }
    return NULL;
}

static void _nonblock(task_queue_t *queue) {
    LOG("Thread: %lu, Start Function: %s\n", pthread_self(), __func__);
    pthread_mutex_lock(&queue->mutex);

    queue->block = 0;

    pthread_mutex_unlock(&queue->mutex);
    pthread_cond_broadcast(&queue->cond);
}


//  使用2级指针，可以不限定任务类型，只要该任务结构体前个8字节为用于指向下一个节点
// 只限定前8个字节为next指针（不限名字）

// 使用一级指针，需要限定任务类型
static inline void _push_task(task_queue_t *queue, void *task) {
    LOG("Thread: %lu, Start Function: %s\n", pthread_self(), __func__);
    // 没有写出next指针名字
    // link 起始地址为task，范围为一个指针大小，即为指向task->next;
    void **link = (void **)task; // **link == &task->next;
    *link = NULL;   // next = NULL;
#if USE_SELF_SPIN
    spinlock_lock(&queue->lock);
#else
    pthread_spin_lock(&queue->lock);
#endif
    *queue->tail = link; /* tail_node->next = task */
    queue->tail = link;
#if USE_SELF_SPIN
    spinlock_unlock(&queue->lock);
#else
    pthread_spin_unlock(&queue->lock);
#endif
    pthread_cond_signal(&queue->cond); // 唤醒一个休眠
}

static inline void *_pop_task(task_queue_t *queue) {
    // LOG("Thread: %lu, Start Function: %s\n", pthread_self(), __func__);
#if USE_SELF_SPIN
    spinlock_lock(&queue->lock);
#else
    pthread_spin_lock(&queue->lock);
#endif
    if (queue->head == NULL) {
#if USE_SELF_SPIN
        spinlock_unlock(&queue->lock);
#else
        pthread_spin_unlock(&queue->lock);
#endif
        return NULL;
    }

    task_t *task;
    task = queue->head;
#if 1
    void **link = (void **)task;
    queue->head = *link;
#else
    queue->head = task->next;
#endif
    if (queue->head == NULL) {
        queue->tail = &queue->head;
    }
#if USE_SELF_SPIN
    spinlock_unlock(&queue->lock);
#else
    pthread_spin_unlock(&queue->lock);
#endif
    return task;
}

static inline void *_get_task(task_queue_t *queue) {
    LOG("Thread: %lu, Start Function: %s\n", pthread_self(), __func__);
    task_t *task;
    /*
        虚假唤醒问题
     */

    while ((task = _pop_task(queue)) == NULL) {
        pthread_mutex_lock(&queue->mutex);
        if (queue->block == 0) {
            pthread_mutex_unlock(&queue->mutex);
            return NULL;
        }

        /*
            cond wait 内部步骤
            1. unlock mutex
            2. 休眠
                ---- _add_task 生产者唤醒
            3. 唤醒
            4. lock mutex
        */
        pthread_cond_wait(&queue->cond, &queue->mutex);
        pthread_mutex_unlock(&queue->mutex);
    }
    return task;
}

static void _taskqueue_destroy(task_queue_t *queue) {
    LOG("Thread: %lu, Start Function: %s\n", pthread_self(), __func__);
    task_t *task;
    while ((task = _pop_task(queue))) {
        free(task);
    }
#if USE_SELF_SPIN
    spinlock_destroy(&queue->lock);
#else
    pthread_spin_destroy(&queue->lock);
#endif
    pthread_cond_destroy(&queue->cond);
    pthread_mutex_destroy(&queue->mutex);
    free(queue);
}

static void *_threadpool_worker(void *arg) {
    LOG("Thread: %lu, Start Function: %s\n", pthread_self(), __func__);
    threadpool_t *pool = (threadpool_t *)arg;
    task_t *task;
    void *ctx;

    while (atomic_load(&pool->quit) == 0) {
        task = (task_t *)_get_task(pool->task_queue);
        if (task == NULL)
            break;
        handler_ptr func = task->func;
        ctx = task->arg;
        free(task);
        func(ctx);
    }

    return NULL;
}

static void _thread_terminate(threadpool_t *pool) {
    LOG("Thread: %lu, Start Function: %s\n", pthread_self(), __func__);
    atomic_store(&pool->quit, 1);
    _nonblock(pool->task_queue);
    int i;
    for (i = 0; i < pool->thread_count; i++) {
        pthread_join(pool->threads[i], NULL);
    }
}

static int _threads_create(threadpool_t *pool, size_t thread_count) {
    LOG("Thread: %lu, Start Function: %s\n", pthread_self(), __func__);
    pthread_attr_t attr;
    int ret;
    ret = pthread_attr_init(&attr);
    if (ret == 0) {
        pool->threads = (pthread_t *)malloc(thread_count * sizeof(pthread_t));
        if (pool->threads) {
            int i = 0;
            for (; i < thread_count; i++) {
                if (0 != pthread_create(&pool->threads[i], &attr, _threadpool_worker, pool)) {
                    break;
                }
            }
            pool->thread_count = i;
            pthread_attr_destroy(&attr);
            if (i == thread_count) {
                return 0;
            }
            _thread_terminate(pool);
            free(pool->threads);
        }
        ret = -1;
    }
    return ret;
}

void threadpool_terminate(threadpool_t *pool) {
    LOG("Thread: %lu, Start Function: %s\n", pthread_self(), __func__);
    atomic_store(&pool->quit, 1);
    _nonblock(pool->task_queue);
    // _thread_terminate(pool);
}

threadpool_t *threadpool_create(int thread_count) {
    LOG("Thread: %lu, Start Function: %s\n", pthread_self(), __func__);
    threadpool_t *pool;
    pool = (threadpool_t *)malloc(sizeof(threadpool_t));
    if (pool) {
        task_queue_t *queue = _taskqueue_create();
        if (queue) {
            pool->task_queue = queue;
            atomic_init(&pool->quit, 0);
            if (0 == _threads_create(pool, thread_count)) {
                return pool;
            }
            _taskqueue_destroy(queue);
        }
        free(pool);
    }
    return NULL;
}

int threadpool_commit(threadpool_t *pool, handler_ptr func, void *arg) {
    LOG("Thread: %lu, Start Function: %s\n", pthread_self(), __func__);
    if (atomic_load(&pool->quit) == 1) {
        return -1;
    }
    task_t *task = (task_t *)malloc(sizeof(task_t));
    if (task == NULL)
        return -1;
    task->func = func;
    task->arg = arg;
    _push_task(pool->task_queue, task);
    return 0;
}

void threadpool_waitdone(threadpool_t *pool) {
    LOG("Thread: %lu, Start Function: %s\n", pthread_self(), __func__);
    int i;
    for (i = 0; i < pool->thread_count; i++) {
        pthread_join(pool->threads[i], NULL);
    }

    _taskqueue_destroy(pool->task_queue);
    free(pool->threads);
    free(pool);
}
#endif